package com.wizo.csb;

import java.io.*;
import java.text.DecimalFormat;
import java.util.Enumeration;
import java.util.Properties;
import java.util.Random;
import java.util.Vector;

import com.wizo.csb.records.Records;
import com.wizo.csb.records.exporter.*;

/**
 * A thread to periodically record the resent period of operations mesages
 */
class UnitRecordThread extends Thread{
	Vector<Thread> _clientThreads;
	public static long _sleepTime = 3000;
	
	public UnitRecordThread(Vector<Thread> clientThreads, long sleepTime){
		_clientThreads = clientThreads;
		_sleepTime = sleepTime;
	}
	
	public void run(){
		boolean allDone;
		do{
			try{
				sleep(_sleepTime);
			}catch(InterruptedException e){
				//do nothing
			}
			
			allDone = true;
			long now = System.currentTimeMillis();
			for(Thread thread:_clientThreads){
				if(thread.getState()!=Thread.State.TERMINATED){
					allDone = false;
				}
				((ClientThread)thread)._measurements.checkEndOfUnit(now);
			}
			
		}while(!allDone);
	}
}

/**
 * A thread to periodically show the status of the experiment, to reassure you that progress is going well.
 * @author ywu70
 */
class StatusThread extends Thread{
	Vector<Thread> _clientThreads;
	public static long _sleepTime = 10000;
	private static long _startTime = 0;
	
	public StatusThread(Vector<Thread> clientThreads, long sleepTime){
		_clientThreads = clientThreads;
		_sleepTime = sleepTime;
	}
	
	public void setStartTime(long start){
		_startTime = start;
	}
	
	public void run(){
		boolean allDone;
		
		do{
			try{
				sleep(_sleepTime);
			}catch(InterruptedException e){
				//do nothing
			}
			
			allDone = true;
			DecimalFormat format = new DecimalFormat("#,###.##");
			long now = System.currentTimeMillis();
			System.out.println("Rum Time(ms): "+format.format(now-_startTime));
			for(Thread thread:_clientThreads){
				System.out.print("[Thread"+((ClientThread)thread)._threadID);
				if(thread.getState()!=Thread.State.TERMINATED){
					allDone = false;
					System.out.print("]");
				}
				else{
					System.out.print("-FINISHED]");
				}
				((ClientThread)thread)._measurements.getOperationInfo();
			}
			
		}while(!allDone);
	}
}


/**
 * A thread for first writing objects or executing transatction to the storage.
 * @author ywu70
 */
class ClientThread extends Thread{
	static Random random = new Random();
	Storage _storage;
	protected Records _measurements;
	String _container;
	boolean _doTransactions;
	Workload _workload;
	int _opCount;
	
	int _opsDone;
	int _threadID;
	Object _workloadState;
	Properties _properties;
	
	/**
	 * Constructor.
	 * 
	 * @param storage the storage implementation to use
	 * @param doTransactions True to do transactions, False to create objects
	 * @param workload the workload to use
	 * @param threadID the id of this thread
	 * @param properties the properties defining the experiment
	 * @param opCount the number of operations(transactions or inserts) to do
	 */
	public ClientThread(Records measurements, Storage storage, String container, boolean doTransactions, 
			Workload workload, int threadID, Properties properties, int opCount){
		//TODO: consider removing threadID
		_measurements = measurements;
		_storage = storage;
		_container = container;
		_doTransactions = doTransactions;
		_workload = workload;
		_threadID = threadID;
		_properties = properties;
		_opCount = opCount;
	}
	
	public int getOpsDone(){
		return _opsDone;
	}
	
	public void run(){
		try{
			_storage.init();
		}catch(StorageException e){
			e.printStackTrace();
			return;
		}
		
		try{
			if(_doTransactions){
				while((_opCount==0)||(_opsDone<_opCount)){
//					System.out.print("Thread"+_threadID);
					if(!_workload.doTransaction(_measurements, _storage, _container))
						break;
					
					_opsDone++;
				}
			}
			else{
				while((_opCount==0)||(_opsDone<_opCount)){
					if(!_workload.doCreate(_measurements, _storage))
						break;
					
					_opsDone++;
				}
			}
		}catch(Exception e){
			e.printStackTrace();
			System.exit(0);
		}
		
		try{
			_storage.cleanup();
		}catch(StorageException e){
			e.printStackTrace();
			return;
		}
	}
}

/**
 * Main class for executing CSB
 * @author ywu70
 *
 */
public class Client {

	public static final String OPERATION_COUNT_PROPERTY = "operationCount";
	public static final String CREATE_OBJECT_COUNT_PROPERTY = "createObjectCount";
	public static final String WORKLOAD_PROPERTY = "workload";
	
	private static void usageMessage(){
		System.out.println("Usage: java com.wizo.csb,Client [options]");
		System.out.println("Options:");
		System.out.println("   -threads n: execute using n threads (default: 1) - can also be specified as the ");
		System.out.println("               \"threadCount\" property using -p");
		System.out.println("   -c: run the create phase of the workload");
		System.out.println("   -t: run the transactions phase of the workload (default)");
		System.out.println("   -storage storageName: specify the name of the Storage to use (default: com.wizo.csb.BasicStorage) - ");
		System.out.println("                         can also be specified as the \"storage\" property using -p");		
		System.out.println("   -P propertyFile: load properties from the given file. Multiple files can be specified, ");
		System.out.println("                    and will be processed in the order specified");
		System.out.println("   -p name=value: specify a property to be passed to the Storage and workload;");
		System.out.println("                  multiple properties can be specified, and override any values in the propertyFile");
		System.out.println("   -s n: show status during run (default: no status) every n ms");
		System.out.println();
		System.out.println("Required properties:");
		System.out.println("   "+WORKLOAD_PROPERTY+": the name of the workload class to use (e.g. com.wizo.csb.workload.CoreWorkload).");
		System.out.println("   bucketCount: the number of the bucket to be operated against.");
		System.out.println("   If bucketBusy is true, threadCount should >= bucketCount. ");
		System.out.println();
	}
	
	private static boolean checkRequiredProperties(Properties properties){
		if(properties.getProperty(WORKLOAD_PROPERTY)==null){
			System.out.println("Missing property: "+WORKLOAD_PROPERTY);
			return false;
		}
		if(properties.getProperty("containerCount")!=null){
			int bucketCountTemp = Integer.parseInt(properties.getProperty("containerCount"));
			for(int i=0; i<bucketCountTemp; i++){
				if(properties.getProperty("container"+i)==null){
					System.out.println("Missing properties: container"+i);
					return false;
				}
			}
		}else{
			System.out.println("Missing properties: containerCount");
			return false;
		}
		if(properties.getProperty("containerBalance")!=null){
			int containerCount = Integer.parseInt(properties.getProperty("containerCount"));
			int threadCount = Integer.parseInt(properties.getProperty("threadCount"));
			boolean containerBalance = Boolean.getBoolean(properties.getProperty("containerBalance"));
			if(containerBalance && containerCount<threadCount){
				System.out.println("Since containerBalance is true, " +
						"threadCount should <= containerCount");
				return false;
			}
		}
		return true;
	}
	
	private static void reportMeasurements(Properties properties, Vector<Records> measurements, 
			long runTime) throws IOException{
		RecordsExporter exporter = null;
		TextRecordsExporter latenciesTRE = null;
		try{
			//If no destination file is provided the results will be wrritten to stdout.
			OutputStream out;
			String exportFile = properties.getProperty("exportFile");
			String latenciesFile = properties.getProperty("latenciesFile");
			
			if(exportFile==null)
				out = System.out;
			else
				out = new FileOutputStream(exportFile);
			
			if(latenciesFile!=null){
				latenciesTRE = new TextRecordsExporter(new FileOutputStream(latenciesFile));
			}
			
			//If no exporter is provided the default TextMeasurementsExporter will be used.
			String exporterClass = properties.getProperty("exporter", 
					"com.wizo.csb.records.exporter.TextMeasurementsExporter");
			try{
				exporter = (RecordsExporter)Class.forName(exporterClass).
					getConstructor(OutputStream.class).newInstance(out);
			}catch(Exception e){
				System.err.println("Could not find exporter " + exporterClass
						+ ", will use default text reporter.");
				e.printStackTrace();
				exporter = new TextRecordsExporter(out);
			}

			DecimalFormat format = new DecimalFormat("#,###.##");
			double bandwidth = 0, aLatency = 0, aOperations = 0;
			long operations = 0;
			boolean getBW = true;
			long now = System.currentTimeMillis();
			int i;
			for(i=0; i<measurements.size(); i++){
				Records mea = measurements.elementAt(i);
				exporter.write("-------------Thread"+i+"-----------------------");
				mea.reportRecords(exporter, now);
				operations += mea.getOperations();
				aLatency += mea.getAverageLatencyPO();
				aOperations += mea.getAverageOperationsPS();
				
				if(getBW&&mea.isGetBW())
					bandwidth += mea.getBandwidth();
				else
					getBW = false;
				
				if(latenciesFile!=null){
					latenciesTRE.write("-------------Thread"+i+"-----------------------");
					mea.reportLatencies(latenciesTRE);
				}
			}
			
			exporter.write("++++++++++++++Summary+++++++++++++++++++++++");
			exporter.write("Operations Count", format.format(operations));
			exporter.write("Run time(ms)", format.format(runTime));
			if(getBW){
				exporter.write("Average Bandwidth(KB/s)", format.format(bandwidth));
			}
			
			exporter.write("Average Latency(ms) Per Operation", format.format(aLatency/i));
			exporter.write("Average Operations Per Second", format.format(aOperations));
			exporter.write("");
			
		}finally{
			if(exporter!=null)
				exporter.close();
			if(latenciesTRE!=null)
				latenciesTRE.close();
		}
	}
	
	public static void main(String[] args){
		String storageName;
		Properties properties = new Properties();
		Properties myFileProps = new Properties();
		boolean doTransactions = true;
		int threadCount = 1;
		boolean status = false;
		
		//parse arguments
		int argIndex = 0;
		
		if(args.length==0){
			usageMessage();
			System.exit(0);
		}
		
		while(args[argIndex].startsWith("-")){
			if(args[argIndex].compareTo("-threads")==0){
				argIndex++;
				if(argIndex>=args.length){
					usageMessage();
					System.exit(0);
				}
				int threads = Integer.parseInt(args[argIndex]);
				properties.setProperty("threadCount", threads+"");
				argIndex++;
			}
			else if(args[argIndex].compareTo("-c")==0){
				doTransactions = false;
				properties.setProperty("doTransactions", doTransactions+"");
				argIndex++;
			}
			else if(args[argIndex].compareTo("-t")==0){
				doTransactions = true;
				properties.setProperty("doTransactions", doTransactions+"");
				argIndex++;
			}
			else if(args[argIndex].compareTo("-storage")==0){
				argIndex++;
				if(argIndex>=args.length){
					usageMessage();
					System.exit(0);
				}
				properties.setProperty("storage", args[argIndex]);
				argIndex++;
			}
			else if(args[argIndex].compareTo("-P")==0){
				argIndex++;
				if(argIndex>=args.length){
					usageMessage();
					System.exit(0);
				}
				String propFile = args[argIndex];
				
				try{
					myFileProps.load(new FileInputStream(propFile));
				}catch(IOException e){
					System.out.println(e.getMessage());
					System.exit(0);
				}
				argIndex++;
			}
			else if(args[argIndex].compareTo("-p")==0){
				argIndex++;
				if(argIndex>=args.length){
					usageMessage();
					System.exit(0);
				}
				int equal = args[argIndex].indexOf("=");
				if(equal<0){
					usageMessage();
					System.exit(0);
				}
				
				String propName = args[argIndex].substring(0, equal);
				String propValue = args[argIndex].substring(equal+1);
				properties.put(propName, propValue);
				
				argIndex++;
			}
			else if(args[argIndex].compareTo("-s")==0){
				argIndex ++;
				if(argIndex>=args.length){
					usageMessage();
					System.exit(0);
				}
				status = true;
				properties.setProperty("status", status+"");
				int statusSleepTime = Integer.parseInt(args[argIndex]);
				properties.setProperty("statusSleepTime", statusSleepTime+"");
				argIndex++;
			}
			else{
				System.out.println("Unknown options "+args[argIndex]);
				usageMessage();
				System.exit(0);
			}
			
			if(argIndex>=args.length)
				break;
		}
		
		//load myFileProps to properties
		for(Enumeration e=myFileProps.propertyNames(); e.hasMoreElements(); ){
			String propName = (String)e.nextElement();
			properties.setProperty(propName, myFileProps.getProperty(propName));
		}
		
		if(!checkRequiredProperties(properties)){
			System.exit(0);
		}
		
		//get number of threads, storage, doTransactions, bucketCount
		threadCount = Integer.parseInt(properties.getProperty("threadCount", "1"));
		storageName = properties.getProperty("storage", "com.wizo.csb.BasicStorage");
		doTransactions = Boolean.parseBoolean(properties.getProperty("doTransactions", "true"));
		boolean containerBalance = Boolean.parseBoolean(
				properties.getProperty("containerBalance", "false"));
		
		//load the workload
		ClassLoader classLoader = Client.class.getClassLoader();
		Workload workload = null;
		
		try{
			Class workloadClass = classLoader.loadClass(properties.getProperty(WORKLOAD_PROPERTY));
			workload = (Workload)workloadClass.newInstance();
		}catch (Exception e){
			e.printStackTrace();
			System.exit(0);
		}
		
		try{
			workload.init(properties);
		}catch(WorkloadException e){
			e.printStackTrace();
			System.exit(0);
		}
		
		System.out.println("Starting test.");
		
		int opCount;
		if(doTransactions)
			opCount = Integer.parseInt(properties.getProperty(OPERATION_COUNT_PROPERTY, "0"));
		else{
			opCount = Integer.parseInt(properties.getProperty(CREATE_OBJECT_COUNT_PROPERTY, "0"));
		}
		
		Vector<Thread> clientThreads = new Vector<Thread>();
		Vector<Records> measurements = new Vector<Records>();
		for(int threadID=0; threadID<threadCount; threadID++){
			Storage storage = null;
			try{
				storage = StorageFactory.newStorage(storageName, properties);
			}catch(UnknownStorageException e){
				System.out.println("Unknown Storage: "+storageName);
				System.exit(0);
			}
			
			Records mea = new Records(properties);
			measurements.add(mea);
			
			String container = null;
			if(containerBalance)
				container = properties.getProperty("container"+threadID);
			
			Thread thread = new ClientThread(mea, storage, container, doTransactions, workload, 
					threadID, properties, opCount/threadCount);
			clientThreads.add(thread);
		}
		
		boolean timeSeries = false;
		if(properties.getProperty(Records.RECORD_TYPE, 
				Records.RECORD_TYPE_DEFAULT).equals(Records.RECORD_TYPE_DEFAULT)){
			long granularity = Long.parseLong(properties.getProperty(
					"timeSeries.granularity", "3000"));
			UnitRecordThread unitRecordThread = new UnitRecordThread(clientThreads, 
					granularity);
			unitRecordThread.start();
			timeSeries = true;
		}
		
		StatusThread statusThread = null;
		if(status){
			statusThread = new StatusThread(clientThreads, 
					Integer.parseInt(properties.getProperty("statusSleepTime")));
			statusThread.start();
		}

		long start = System.currentTimeMillis();
		if(status)
			statusThread.setStartTime(start);
		if(timeSeries){
			for(Records mea:measurements){
				mea.setStart(start);
			}
		}
		
		for(Thread thread:clientThreads){
			thread.start();
		}
		for(Thread thread:clientThreads){
			try{
				thread.join();
			}catch(InterruptedException e){
				
			}
		}
		long end = System.currentTimeMillis();
		
		try{
			workload.cleanup();
		}catch(WorkloadException e){
			e.printStackTrace();
			System.exit(0);
		}

		System.out.println("Test finished.");
		try {
			reportMeasurements(properties, measurements, end-start);
		} catch (IOException e) {
			e.printStackTrace();
		}
		
		System.exit(0);
	}
}
